#ifndef _M_CHANNEL_H_
#define _M_CHANNEL_H_

#include "../mqthird/include/muduo/proto/dispatcher.h"
#include "../mqthird/include/muduo/proto/codec.h"
#include <muduo/net/TcpConnection.h>

#include "../mqcommon/threadpool.hpp"
#include "../mqcommon/mqProto.pb.h"
#include "../mqcommon/mqlink.pb.h"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/helper.hpp"

#include "consumer.hpp"
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <unordered_map>

namespace MyRabbitMQ
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicPushResponsePtr = std::shared_ptr<protoMQ::basicPushResponse>;
    using basicCommonResponsePtr = std::shared_ptr<protoMQ::basicCommonResponse>;

    //信道类
    class Channel
    {

    public:
        using ptr = std::shared_ptr<Channel>;

        Channel(const muduo::net::TcpConnectionPtr& conn
        ,const ProtobufCodecPtr& codec)
        :m_id(UuidHelper::uuid()),m_conn(conn),m_codec(codec){}

        ~Channel(){this->queueCancelConsume();};

        //开启信道
        bool openChannel()
        {

            protoMQ::openChannelRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_request_id(rid);
            req.set_channel_id(m_id);
            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            auto resp  = this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return resp->ok();
        }

        //关闭信道
        void closeChannel()
        {
            protoMQ::closeChannelRequest req;
            std::string rid = UuidHelper::uuid();
            req.set_request_id(rid);
            req.set_channel_id(m_id);
            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return;
        }
        
        //声明交换机
        bool declareExchange(const std::string &name, protoMQ::ExchangeType type, 
        bool isDurable, bool isAutoDelete, const google::protobuf::Map<std::string, std::string> &args)
        {
            protoMQ::declareExchangeRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_exchange_name(name);
            req.set_type(type);
            req.set_isdurable(isDurable);
            req.set_isautodelete(isAutoDelete);
            google::protobuf::Map<std::string, std::string> tmp_args = args;
            req.mutable_args()->swap(tmp_args);

            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            auto resp  = this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return resp->ok();
        }

        //删除交换机
        void deleteExchange(const std::string &name)
        {
            protoMQ::deleteExchangeRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_exchange_name(name);

            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return;
        }

        //声明队列
        bool declareQueue(const std::string &name
        ,bool isUnique
        ,bool isDurable
        ,bool isAutoDelete
        ,const google::protobuf::Map<std::string, std::string> &args)
        {
            protoMQ::declareQueueRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_queue_name(name);
            req.set_isdurable(isDurable);
            req.set_isunique(isUnique);
            req.set_isautodelete(isAutoDelete);
            google::protobuf::Map<std::string, std::string> tmp_args = args;
            req.mutable_args()->swap(tmp_args);

            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            auto resp  = this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return resp->ok();
        }

        //删除队列
        void deleteQueue(const std::string& name)
        {
            protoMQ::deleteQueueRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_queue_name(name);
  
            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return;
        }

        //声明绑定
        bool bind(const std::string& ename, const std::string& qname, const std::string& bindingKey)
        {
            protoMQ::bindRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_queue_name(qname);
            req.set_exchange_name(ename);
            req.set_binding_key(bindingKey);

            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            auto resp  = this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return resp->ok();
        }

        //解除绑定
        void unBind(const std::string& ename, const std::string& qname)
        {
            protoMQ::unBindRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_queue_name(qname);
            req.set_exchange_name(ename);

            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return;
        }

        //队列订阅
        bool queueConsume(const std::string& qname, 
        const std::string& consumer_tag, 
        bool isAutoAck,
        const MyRabbitMQ::NSConsumer::Callback& cb)
        {
            if(this->m_consumer.get() != nullptr){ return false; }

            protoMQ::queueConsumeRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_queue_name(qname);
            req.set_consumer_tag(consumer_tag);
            req.set_isautoack(isAutoAck);

            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            auto resp  = this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;

            if(!resp->ok())
            {
                return false;
            }
            
            //实例化消费者对象
            m_consumer = std::make_shared<Consumer>(consumer_tag,qname,isAutoAck,cb);
            return true;
        }

        //队列取消订阅
        void queueCancelConsume()
        {
            if(this->m_consumer.get() == nullptr) return;

            protoMQ::queueCancelConsumeRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_queue_name(this->m_consumer->m_subscriptionQueue);
            req.set_consumer_tag(this->m_consumer->m_name);

            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;

            //重置消费者对象
            this->m_consumer.reset();
            return;
        }


        //发布消息
        void basicPublish(const std::string &ename, const protoMQ::BaseProperties *base,const std::string &body)
        {
            protoMQ::basicPublishRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_body(body);
            req.set_exchange_name(ename);

            if(base != nullptr)
            {
                protoMQ::BaseProperties bp;
                req.mutable_properties()->set_id(base->id());
                req.mutable_properties()->set_mode(base->mode());
                req.mutable_properties()->set_routing_key(base->routing_key());
            }

            this->m_codec->send(m_conn,req);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;
            return;
        }
        
        //确认消息
        void basicAck(const std::string& message_id)
        {
            if(this->m_consumer.get() == nullptr) return;

            protoMQ::basicAckRequest req;

            std::string rid = UuidHelper::uuid();

            req.set_request_id(rid);
            req.set_channel_id(m_id);
            req.set_queue_name(this->m_consumer->m_subscriptionQueue);
            req.set_message_id(message_id);

            this->m_codec->send(m_conn,req);

            LOG(logLevel::DEBUG) << "信道" << this->m_id << "发送了" << __func__ << "请求" << std::endl;
            this->waitResponse(rid);
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "收到了" << __func__ << "响应" << std::endl;

            return;
        }

        //向hash添加基础响应
        void pushBasicCommonResponse(const basicCommonResponsePtr& resp)
        {
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "准备执行" << std::endl;
            std::unique_lock<std::mutex> lock(m_mtx);
            this->m_responses.insert(std::make_pair(resp->response_id(),resp));
            this->m_cv.notify_all();
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "执行完毕" << std::endl;
            return;
        }

        //消息推送响应处理
        void handleBasicPushResponse(const basicPushResponsePtr& resp)
        {
            // std::unique_lock<std::mutex> lock(m_mtx); // 这里是没有必要加锁的,也肯定不能加锁的,因为在callback里边一旦有其他操作涉及到加锁就会死锁,比如basicAck里边的waitResponse
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "准备执行" << std::endl;
            if(this->m_consumer.get() == nullptr) {return;}
            if(this->m_consumer->m_name != resp->consumer_tag()) {return;}
            m_consumer->m_callback(resp->consumer_tag(),&resp->properties(),resp->body());
            LOG(logLevel::DEBUG) << "信道" << this->m_id << "执行完毕" << std::endl;
            return;
        }

        std::string getChannelId(){return m_id;}

    private:
        std::string m_id;                       //信道id
        muduo::net::TcpConnectionPtr m_conn;    //信道所属tcp连接
        ProtobufCodecPtr m_codec;               //protobuf协议处理器, 处理序列化数据
        Consumer::ptr m_consumer;               //信道对应的消费者(这里的消费者是指订阅客户端, 所以该字段
                                                //不一定有效)
        
        std::unordered_map<std::string, basicCommonResponsePtr> m_responses; //响应队列
        std::mutex m_mtx;
        std::condition_variable m_cv;

        //等待响应
        basicCommonResponsePtr waitResponse(const std::string& rid)
        {
            std::unique_lock<std::mutex> lock(m_mtx);
            m_cv.wait(lock,[&rid, this](){return this->m_responses.find(rid) != m_responses.end();});
            basicCommonResponsePtr ret = m_responses[rid];
            m_responses.erase(rid);
            return ret;
        }
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;

        ChannelManager(){}

        //创建信道
        Channel::ptr createChannel(const muduo::net::TcpConnectionPtr& conn,const ProtobufCodecPtr& codec)
        {
            std::unique_lock<std::mutex>(m_mtx);
            auto ptr = std::make_shared<Channel>(conn, codec);
            m_channels.insert(std::make_pair(ptr->getChannelId(),ptr));
            return ptr;
        }

        //删除信道
        void deleteChannel(const std::string& cid)
        {
            std::unique_lock<std::mutex>(m_mtx);
            m_channels.erase(cid);
        }

        //获取特定信道
        Channel::ptr getChannnel(const std::string& cid)
        {
            std::unique_lock<std::mutex>(m_mtx);
            auto it = m_channels.find(cid);
            
            if(it == m_channels.end()) return Channel::ptr();

            return it->second;
        }

    private:
        std::mutex m_mtx;
        std::unordered_map<std::string, Channel::ptr> m_channels;
    };
}

#endif